use std::sync::Arc;
use std::time::Duration;

use async_listen::{error_hint, ListenExt};
use async_std::net::{TcpListener, TcpStream};
use xactor::{Actor, Addr, Context, StreamHandler};

use crate::prelude::*;

use super::tcpstream::{ListenDelegate, TcpStreamActor};

pub struct TcpListenerActor {
    node_type: String,
    node_name: String,
    addr: Option<Addr<Self>>,
    listen_port: u16,
    message_register: Arc<MessageRegister>,
    listen_delegate: Option<Arc<dyn ListenDelegate>>,
}

impl TcpListenerActor {
    pub fn new(
        node_type: &str,
        node_name: &str,
        listen_port: u16,
        message_register: Arc<MessageRegister>,
        listen_delegate: Option<Arc<dyn ListenDelegate>>,
    ) -> Self {
        Self {
            node_type: node_type.into(),
            node_name: node_name.into(),
            addr: None,
            listen_port,
            message_register,
            listen_delegate,
        }
    }
}

#[async_trait::async_trait]
impl Actor for TcpListenerActor {
    fn name(&self) -> &str {
        "TcpListenerActor"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> crate::Result<()> {
        info!("tcp listener actor:{} started", ctx.actor_id());

        self.addr = Some(ctx.address());
        let listener = Box::new(
            TcpListener::bind(("0.0.0.0", self.listen_port))
                .await
                .unwrap(),
        );
        ctx.add_stream(
            Box::leak(listener)
                .incoming()
                .log_warnings(|err| {
                    error!("Accept error: {}. Sleeping 0.5s. {}", err, error_hint(&err))
                })
                .handle_errors(Duration::from_millis(500)),
        );
        Ok(())
    }
}

#[async_trait::async_trait]
impl StreamHandler<TcpStream> for TcpListenerActor {
    async fn handle(&mut self, _ctx: &mut Context<Self>, stream: TcpStream) {
        let node_type = self.node_type.clone();
        let node_name = self.node_name.clone();
        let message_register = self.message_register.clone();
        let listen_delegate = self.listen_delegate.clone();
        spawn(async move {
            let _ = stream.set_nodelay(true);
            let mut framed_write =
                FramedWrite::new(stream.clone().compat(), HoneybeeMsgCodec::new());
            let mut framed_read = FramedRead::new(stream.clone().compat(), HoneybeeMsgCodec::new());
            let msg: TcpRegister = match framed_read.read_msg().await {
                Ok(msg) => msg,
                Err(err) => {
                    error!("{}@{} read register msg err:{}", node_name, node_type, err);
                    return;
                }
            };

            if let Err(err) = framed_write
                .send_ext(TcpRegister::new(&node_type, &node_name))
                .await
            {
                error!("{}@{} send register msg err:{}", node_name, node_type, err);
                return;
            }

            let remote_node_type = msg.node_type;
            let remote_node_name = msg.node_name;

            if let Err(err) = TcpStreamActor::new(
                &remote_node_type,
                &remote_node_name,
                framed_write,
                framed_read,
                message_register.clone(),
                listen_delegate.clone(),
            )
            .start()
            .await
            {
                error!(
                    "{}@{} new tcp stream actor err:{}",
                    node_type, node_name, err
                );
                return;
            }
            info!(
                "{}@{} -> {}@{} connected success",
                remote_node_name, remote_node_type, node_name, node_type
            );
        });
    }

    async fn finished(&mut self, ctx: &mut Context<Self>) {
        info!("TcpListener stoped");
        let _ = ctx.stop(None);
    }
}
